﻿using System;
using System.Net;
using System.Net.Sockets;
using Force.Crc32;
using System.Collections.Concurrent;
using System.Collections.Generic;
using System.Threading.Tasks;
using System.Threading;
using GSE.Network;


namespace GSE.Network.UDP
{
    public class UDPTransport : Transport
    {
        const int SeqDedupDepth = 10;
        protected OptionContext optCtx;
        private Socket socket;
        private UInt16 nextPacketSN;
        protected Task recvTask;
        protected CancellationTokenSource keepAliveTaskCancellationTokenSource;
        protected Task keepAliveTask;
        protected string remoteAddress;
        protected Mutex sendMutex = new Mutex();
        protected Mutex readMutex = new Mutex();
        // 控制数据队列: ack、reset.
        protected BlockingCollection<Packet> abQueue;
        // 应用数据队列.
        protected BlockingCollection<Packet> dataQueue;
        // 下次调用ReadData时的超时时间. 单位: 毫秒.
        protected int nextReadTimeout;

        protected Packet[] deDupSeqList = new Packet[SeqDedupDepth];

        protected int deDupSeqListNextIndex = 0;

        protected CancellationTokenSource netErrCancellationTokenSource;

        protected bool normalClosed = false;

        protected ILogger logger
        {
            get { return Logger.Default; }
        }

        public void ReceiveBufferSize(Int32 size)
        {
            socket.ReceiveBufferSize = size;
        }

        public Int32 ReceiveBufferSize()
        {
            return socket.ReceiveBufferSize;
        }

        public void SendBufferSize(Int32 size)
        {
            socket.SendBufferSize = size;
        }

        public Int32 SendBufferSize()
        {
            return socket.SendBufferSize;
        }

        public UDPTransport(string remoteAddr, OptionContext optCtx)
        {
            var portions = remoteAddr.Split(':');
            if (portions.Length != 2)
            {
                throw new Exception("Invalid remote address: " + remoteAddr);
            }

            remoteAddress = remoteAddr;
            var ips = Dns.GetHostAddresses(portions[0]);
            socket = new Socket(AddressFamily.InterNetwork, SocketType.Dgram, ProtocolType.Udp);
            EndPoint ep = new IPEndPoint(ips[0], int.Parse(portions[1]));
            nextPacketSN = 1;
            socket.Connect(ep);
            this.optCtx = optCtx;
            abQueue = new BlockingCollection<Packet>(10000);
            dataQueue = new BlockingCollection<Packet>(10000);
        }

        /// <summary>
        /// 设置CancellationTokenSource后,当客户端在等待网络IO时,如果网络有错误则立即退出等待操作.
        /// </summary>
        /// <param name="cts">CancellationTokenSource</param>
        public void SetNetErrCancellationTokenSource(CancellationTokenSource cts)
        {
            netErrCancellationTokenSource = cts;
        }

        public void Start()
        {
            recvTask = Task.Run(() => recv());
        }

        public void Close()
        {
            sendMutex.WaitOne();
            _close();
            sendMutex.ReleaseMutex();
        }

        public void Close(bool force)
        {
            if (!force)
            {
                sendMutex.WaitOne();
            }
            _close();
            if (!force)
            {
                sendMutex.WaitOne();
            }
        }

        protected void _close()
        {
            normalClosed = true;
            logger.Debug($"Closing client of {remoteAddress}.");
            if (keepAliveTaskCancellationTokenSource != null)
            {
                keepAliveTaskCancellationTokenSource.Cancel();
            }
            socket.Close();
            recvTask.Wait(3000);
            logger.Debug($"recv task status:{recvTask.Status}");
            if (keepAliveTask != null)
            {
                keepAliveTask.Wait(3000);
                logger.Debug($"keeepalive task status:{keepAliveTask.Status}");
            }
            logger.Debug($"Closed client of {remoteAddress}.");
        }

        protected void recv()
        {
            int n = 0;
            while (true)
            {
                byte[] udpPacketRaw = new byte[Packet.PacketSize];
                bool hasException = false;
                try
                {
                    n = socket.Receive(udpPacketRaw);
                }
                catch (ObjectDisposedException)
                {
                    logger.Debug("Client closed.");
                    hasException = true;
                    return;
                }
                catch (Exception ex)
                {
                    if (!socket.Connected && !normalClosed)
                        logger.Warn("Client socket err: " + ex.ToString());
                    hasException = true;
                    return;
                }
                finally
                {
                    if (hasException)
                    {
                        if (netErrCancellationTokenSource != null)
                        {
                            netErrCancellationTokenSource.Cancel(true);
                        }
                    }
                }

                if (n < udpPacketRaw.Length)
                {
                    Array.Resize(ref udpPacketRaw, n);
                }
                Packet udpPacket;
                try
                {
                    udpPacket = new Packet(udpPacketRaw, remoteAddress);
                }
                catch (Exception ex)
                {
                    logger.Warn($"packet decode exception: {ex.Message} {ex.StackTrace}");
                    continue;
                }
                logger.Debug($"received remote UDP packet, type: {udpPacket.GetPacketType()}, seq: {udpPacket.GetSeqNo()}");
                switch (udpPacket.GetPacketType())
                {
                    case PacketType.PacketTypeAck:
                    case PacketType.PacketTypeReset:
                    case PacketType.PacketTypeClose:
                        abQueue.Add(udpPacket);
                        break;
                    case PacketType.PacketTypePing:
                        break;
                    default:
                        dataQueue.Add(udpPacket);
                        break;
                }
            }
        }

        protected UInt16 genPacketSN()
        {
            UInt16 n = nextPacketSN;
            nextPacketSN += 1;

            if (nextPacketSN > Packet.MaxPacketSeqNo)
            {
                nextPacketSN = 1;

            }

            return n;
        }

        public void KeepAlive()
        {
            var heartbeat = Packet.NewHeartbeat();
            keepAliveTaskCancellationTokenSource = new CancellationTokenSource();
            keepAliveTask = Task.Run(() =>
            {
                var timer = new System.Timers.Timer(2000);
                timer.Elapsed += (s, e) =>
                {
                    if (keepAliveTaskCancellationTokenSource.Token.IsCancellationRequested)
                    {
                        return;
                    }
                    sendMutex.WaitOne();
                    try
                    {
                        socket.Send(heartbeat);
                        sendMutex.ReleaseMutex();
                    }
                    catch (Exception ex)
                    {
                        sendMutex.ReleaseMutex();
                        throw ex;
                    }

                };
                timer.Start();

            }, keepAliveTaskCancellationTokenSource.Token);
        }

        public int SendData(byte[] data)
        {
            return mutexSend(data, false);
        }

        public int SendDataWithAck(byte[] data)
        {
            return mutexSend(data, true);
        }

        protected int mutexSend(byte[] data, bool ack)
        {
            try
            {
                sendMutex.WaitOne();
                var n = send(data, ack);
                sendMutex.ReleaseMutex();
                return n;
            }
            catch (Exception ex)
            {
                sendMutex.ReleaseMutex();
                throw ex;
            }
        }

        public int ReadData(byte[] buf)
        {
            throw new Exception(" \"int ReadData(byte[] buf)\" not allowd for UDP transport.");
        }

        public byte[] ReadData()
        {
            try
            {
                readMutex.WaitOne();
                var data = readData();
                readMutex.ReleaseMutex();
                return data;
            }
            catch (Exception ex)
            {
                readMutex.ReleaseMutex();
                throw ex;
            }
        }

        protected int send(byte[] data, bool ack)
        {
            if (Packet.MaxDataLength < (UInt32)data.Length)
            {
                throw new Exception($"data length {data.Length} exceeded {Packet.MaxDataLength}.");
            }
            byte[] header = new byte[2];
            byte[] dataBytes;

            var dataCRC = Crc32Algorithm.Compute(data);

            PacketType packetType;
            int totalSend = 0;

            while (totalSend < data.Length)
            {
                byte[] udpPacket = new byte[Packet.PacketSize];
                UInt16 headerInt = (UInt16)Packet.CurrentProtoVerId << 14;
                if (ack)
                {
                    headerInt += (UInt16)1 << 13;
                }

                if (totalSend == 0)
                {
                    // 减去首包头部10个字节.
                    if (Packet.PacketSize - 10 >= data.Length)
                    {
                        packetType = PacketType.PacketTypeHeadWithTail;
                        dataBytes = data;
                    }
                    else
                    {
                        packetType = PacketType.PacketTypeHead;
                        dataBytes = new byte[Packet.PacketSize - 10];
                        Array.Copy(data, dataBytes, dataBytes.Length);
                    }
                }
                else
                {
                    // 减去头部2字节.计算是否能容下剩余字节数.
                    if (Packet.PacketSize - 2 > data.Length - totalSend)
                    {
                        packetType = PacketType.PacketTypeTail;
                        dataBytes = new byte[data.Length - totalSend];
                        Array.Copy(data, totalSend, dataBytes, 0, dataBytes.Length);
                    }
                    else
                    {
                        packetType = PacketType.PacketTypePartial;
                        dataBytes = new byte[Packet.PacketSize - 2];
                        Array.Copy(data, totalSend, dataBytes, 0, dataBytes.Length);
                    }

                }

                headerInt += (UInt16)((UInt16)packetType << 10);
                var seqNo = genPacketSN();
                headerInt += seqNo;

                header = BitConverter.GetBytes(headerInt);
                Array.Reverse(header);
                Array.Copy(header, udpPacket, header.Length);
                int offsetT = header.Length;

                if (packetType == PacketType.PacketTypeHead || packetType == PacketType.PacketTypeHeadWithTail)
                {
                    // put data length
                    var dataLengthBytes = BitConverter.GetBytes((UInt32)data.Length);
                    Array.Reverse(dataLengthBytes);
                    Array.Copy(dataLengthBytes, 0, udpPacket, offsetT, dataLengthBytes.Length);
                    offsetT += 4;

                    // put crc32
                    var crcBytes = BitConverter.GetBytes(dataCRC);
                    Array.Reverse(crcBytes);
                    Array.Copy(crcBytes, 0, udpPacket, offsetT, crcBytes.Length);
                    offsetT += 4;
                }
                Array.Copy(dataBytes, 0, udpPacket, offsetT, dataBytes.Length);
                offsetT += dataBytes.Length;

                UInt16 sendRetries = 3;

                if (offsetT < udpPacket.Length)
                {
                    Array.Resize(ref udpPacket, offsetT);
                }

                for (; sendRetries > 0; sendRetries--)
                {
                    try
                    {
                        var sendN = socket.Send(udpPacket);
                        if (sendN != udpPacket.Length)
                        {
                            throw new Exception($"Sent incomplete UDP packet, expects sent {udpPacket.Length} bytes, but have {sendN}");
                        }
                        if (ack)
                        {
                            waitAck(seqNo);
                        }
                        totalSend += dataBytes.Length;
                        logger.Debug($"Total data {data.Length} bytes, already sent {totalSend} bytes.");
                        break;
                    }
                    catch (Exception ex)
                    {
                        if (sendRetries == 1)
                        {
                            throw new Exception("wait ack error.", ex);
                        }
                        logger.Debug($"try resend, previous exception message: {ex.ToString()}");
                        continue;
                    }

                }
            }


            return totalSend;
        }

        // 设置下次调用ReadData的超时时间.
        public void SetReadTimeout(int timeout)
        {
            nextReadTimeout = timeout;
        }

        protected byte[] readData()
        {
            var payloadPackets = new List<byte[]>();

            UInt32 payloadTotalLength = 0;
            byte[] payload;
            bool dataHeadReceived = false;
            UInt32 dataCRC = 0;
            UInt32 payloadExpectedLength = 0;
            Packet udpPacket;
            READ:
            while (true)
            {
                if (nextReadTimeout > 0)
                {
                    var success = dataQueue.TryTake(out udpPacket, nextReadTimeout);
                    if (!success)
                    {
                        throw new Exception("Read data timedout.");
                    }
                }
                else
                {
                    udpPacket = dataQueue.Take();
                }

                if (udpPacket.GetPacketType() == PacketType.PacketTypeReset)
                {
                    throw new ConnectionResetException(remoteAddress);
                }
                if (udpPacket.GetPacketType() == PacketType.PacketTypeClose)
                {
                    throw new ConnectionCloseException(remoteAddress);
                }

                for (int i = 0; i < deDupSeqList.Length; i++)
                {
                    if (deDupSeqList[i] == null)
                    {
                        break;
                    }
                    if (deDupSeqList[i].GetSeqNo() == udpPacket.GetSeqNo())
                    {
                        logger.Debug($"packet may be duplicated, dropped, seq[{udpPacket.GetSeqNo()}], remote_addr[{remoteAddress}]");
                        goto READ;
                    }
                }
                deDupSeqList[deDupSeqListNextIndex] = udpPacket;

                deDupSeqListNextIndex += 1;

                if (deDupSeqListNextIndex >= SeqDedupDepth)
                {
                    deDupSeqListNextIndex = 0;

                }

                payload = udpPacket.GetPayload();
                if (udpPacket.IsHead())
                {
                    if (dataHeadReceived)
                    {
                        throw new Exception($"data broken: data head already received! packet seq[{udpPacket.GetSeqNo()}], packet type[{udpPacket.GetPacketType()}]");
                    }
                    dataCRC = udpPacket.GetCRC();
                    payloadExpectedLength = udpPacket.GetDataLength();
                    dataHeadReceived = true;
                }
                else if (udpPacket.IsTail())
                {
                    if (!dataHeadReceived && udpPacket.GetPacketType() != PacketType.PacketTypeHeadWithTail)
                    {
                        logger.Warn($"head data is expected firstly, but tail received. data dropped! packet seq[{udpPacket.GetSeqNo()}], packet type[{udpPacket.GetPacketType()}], remote addr[{remoteAddress}]");
                        continue;

                    }
                }
                else if (!dataHeadReceived)
                {
                    logger.Warn($"head data is expected firstly, data dropped! packet seq[{udpPacket.GetSeqNo()}], packet type[{udpPacket.GetPacketType()}], , remote addr[{remoteAddress}]");
                    continue;
                }

                if (payload.Length > 0)
                {
                    payloadPackets.Add(payload);
                    payloadTotalLength += (UInt32)payload.Length;
                    logger.Debug($"progress: total:{payloadExpectedLength} received: {payloadTotalLength} remaining:{payloadExpectedLength - payloadTotalLength} percent:{(float)payloadTotalLength / (float)payloadExpectedLength}");
                }

                if (payloadTotalLength > payloadExpectedLength)
                {
                    throw new Exception($"data broken: read data length incrrect! expected length[{payloadExpectedLength}], crrently read bytes[{payloadTotalLength}]");
                }
                if (udpPacket.IsTail())
                {
                    break;
                }
            }
            byte[] data = new byte[payloadTotalLength];
            int offset = 0;
            payloadPackets.ForEach(delegate (byte[] p)
            {
                Array.Copy(p, 0, data, offset, p.Length);
                offset += p.Length;
            });
            var checkedCRC = Crc32Algorithm.Compute(data);
            if (checkedCRC != dataCRC)
            {
                throw new Exception($"read bad data. expected_length:{payloadExpectedLength}, expected_crc: {dataCRC}, actual_length: {payloadTotalLength}, actual_crc: {checkedCRC}");
            }
            return data;
        }

        protected void waitAck(UInt16 sn)
        {
            Packet ackPacket;
            logger.Debug($"waiting ack [{sn}]");
            TryGetAck:
            {
                var success = abQueue.TryTake(out ackPacket, 2000);
                if (!success)
                {
                    throw new Exception($"wait ack for seq[{sn}] timedout.");
                }
                else if (ackPacket.GetPacketType() == PacketType.PacketTypeReset)
                {
                    throw new ConnectionResetException(remoteAddress);
                }
                if (ackPacket.GetSeqNo() != sn)
                {
                    abQueue.Add(ackPacket);
                    logger.Debug($"seq[{sn}] received un-expected ack {ackPacket.GetSeqNo()}, release it.");
                    Thread.Sleep(1);
                    goto TryGetAck;
                }
            }
            logger.Debug($"seq[{sn}] received ack {ackPacket.GetSeqNo()}");
        }

    }


}
